package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.VisitorStats;
import com.atguigu.realtime.util.KKutil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;

import static com.atguigu.realtime.common.Constant.*;

public class DwsVisitorStatsApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsVisitorStatsApp().init(4001,1,"DwsVisitorStatsApp","DwsVisitorStatsApp",
                TOPIC_DWD_PAGE, TOPIC_DWM_UV,TOPIC_DWM_UJ);
    }
    @Override
    protected void run(StreamExecutionEnvironment env,
                       HashMap<String, DataStreamSource<String>> topicToStream) {



        //1.把多个流union成一个流
        DataStream<VisitorStats> vsStream = parseAndUnionOne(topicToStream);

        //2.开窗聚合
        SingleOutputStreamOperator<VisitorStats> aggregatedStream = windowAndAggregate(vsStream);
        aggregatedStream.print();

        //3.数据写入到Clickhouse


    }

    private SingleOutputStreamOperator<VisitorStats> windowAndAggregate(DataStream<VisitorStats> vsStream) {
       return vsStream
                .keyBy(vs ->vs.getVc()+"_"+vs.getCh()+"_"+vs.getAr()+"_"+vs.getIs_new())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<VisitorStats>() {
                            @Override
                            public VisitorStats reduce(VisitorStats vs1, VisitorStats vs2) throws Exception {
                                vs1.setUv_ct(vs1.getUv_ct() + vs2.getUv_ct());
                                vs1.setPv_ct(vs1.getPv_ct() + vs2.getPv_ct());
                                vs1.setSv_ct(vs1.getSv_ct() + vs2.getSv_ct());
                                vs1.setUj_ct(vs1.getUj_ct() + vs2.getUj_ct());
                                vs1.setDur_sum(vs1.getDur_sum() + vs2.getDur_sum());
                                return vs1;
                            }
                        },
                        //获取窗口信息
                        new ProcessWindowFunction<VisitorStats, VisitorStats, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<VisitorStats> elements,
                                                Collector<VisitorStats> out) throws Exception {
                                VisitorStats vs = elements.iterator().next();
                                String stt = KKutil.toDateTime(ctx.window().getStart());
                                String edt = KKutil.toDateTime(ctx.window().getEnd());

                                vs.setStt(stt);
                                vs.setEdt(edt);
                                // 把ts设置为当前系统时间
                                vs.setTs(System.currentTimeMillis());
                                out.collect(vs);
                            }
                        }
                );

    }

    private DataStream<VisitorStats> parseAndUnionOne(HashMap<String, DataStreamSource<String>> topicToStream) {

        //1.计算PV  he  dur_sum
        SingleOutputStreamOperator<VisitorStats> pvAndDtStream = topicToStream
                .get(TOPIC_DWD_PAGE)
                .map(line -> {
                    JSONObject obj = JSON.parseObject(line);
                    JSONObject common = obj.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String is_new = common.getString("is_new");
                    Long ts = obj.getLong("ts");
                    Long dur_sum = obj.getJSONObject("page").getLong("during_time");
                    return new VisitorStats(
                            "", "",
                            vc, ch, ar, is_new,
                            0L, 1L, 0L, 0L, dur_sum,
                            ts);
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((vs,ts)->vs.getTs())
                )
                ;

        //2.计算UV
        SingleOutputStreamOperator<VisitorStats> uvStream = topicToStream
                .get(TOPIC_DWM_UV)
                .map(line -> {
                    JSONObject obj = JSON.parseObject(line);
                    JSONObject common = obj.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String is_new = common.getString("is_new");
                    Long ts = obj.getLong("ts");
                    return new VisitorStats(
                            "", "",
                            vc, ch, ar, is_new,
                            1L, 0L, 0L, 0L, 0L,
                            ts);
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((vs,ts)->vs.getTs())
                );

        //3.计算UJ
        SingleOutputStreamOperator<VisitorStats> ujStream = topicToStream
                .get(TOPIC_DWM_UJ)
                .map(line -> {
                    JSONObject obj = JSON.parseObject(line);
                    JSONObject common = obj.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String is_new = common.getString("is_new");
                    Long ts = obj.getLong("ts");
                    return new VisitorStats(
                            "", "",
                            vc, ch, ar, is_new,
                            0L, 0L, 0L, 1L, 0L,
                            ts);
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((vs,ts)->vs.getTs())
                );
        //4.计算sv
        SingleOutputStreamOperator<VisitorStats> svStream = topicToStream
                .get(TOPIC_DWD_PAGE)
                .flatMap(new FlatMapFunction<String, VisitorStats>() {
                    @Override
                    public void flatMap(String value, Collector<VisitorStats> out) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        if (obj.getJSONObject("page").getString("last_page_id") == null) {
                            JSONObject common = obj.getJSONObject("common");
                            String vc = common.getString("vc");
                            String ch = common.getString("ch");
                            String ar = common.getString("ar");
                            String is_new = common.getString("is_new");
                            Long ts = obj.getLong("ts");
                            out.collect(
                                    new VisitorStats(
                                            "", "",
                                            vc, ch, ar, is_new,
                                            0L, 0L, 1L, 0L, 0L,
                                            ts));
                        }
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((vs,ts)->vs.getTs())
                );
        return pvAndDtStream.union(uvStream,ujStream,svStream);

    }
}
